/**
 * Copyright (c) 2013-2022 Nikita Koksharov
 * <p>
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * <p>
 * http://www.apache.org/licenses/LICENSE-2.0
 * <p>
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.redisson;

import io.netty.util.Timeout;
import org.redisson.api.RFuture;
import org.redisson.client.RedisTimeoutException;
import org.redisson.client.codec.LongCodec;
import org.redisson.client.protocol.RedisCommands;
import org.redisson.client.protocol.RedisStrictCommand;
import org.redisson.command.CommandAsyncExecutor;
import org.redisson.misc.CompletableFutureWrapper;
import org.redisson.pubsub.LockPubSub;

import java.util.Arrays;
import java.util.Collections;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;

/**
 * 分布式锁，重入锁
 */
public class RedissonLock extends RedissonBaseLock {

    /**
     * 看门狗超时时间
     * 设置内部锁的租约时间，租约时间是指锁的持有时间。即在获得锁后可以在多长时间内保持锁而不释放。
     * 当一个线程获取一个锁时，它会计算租约时间，如果在租约时间内没有释放锁，那么锁将会自动释放，那么锁将自动释放，从而允许其他线程获取该锁
     * 租约时间的设置可以控制锁的持有时间，从而控制并发访问资源的竞争情况
     */
    protected long internalLockLeaseTime;

    /**
     * LockPubSub是一个用于发布和订阅消息的类。这个成员变量是保护级别的，
     * 说明它只能在当前类或其子类中访问。使用LockPubSub可以帮助实现线程安全的消息发布和订阅机制。
     */
    protected final LockPubSub pubSub;

    /**
     * 声明一个CommandAsyncExecutor类型的成员变量，该变量用于异步执行命令。
     * CommandAsyncExecutor是一个用于异步执行命令的接口或类，具体的实现可能包含在其它部分的代码中。
     * 该成员变量被声明为final，意味着它在初始化之后不能被重新赋值。
     */
    final CommandAsyncExecutor commandExecutor;

    public RedissonLock(CommandAsyncExecutor commandExecutor, String name) {
        super(commandExecutor, name);
        this.commandExecutor = commandExecutor;
        // 看门狗时间
        this.internalLockLeaseTime = getServiceManager().getCfg().getLockWatchdogTimeout();
        this.pubSub = commandExecutor.getConnectionManager().getSubscribeService().getLockPubSub();
    }


    /**
     * 用于发布pubsub发布消息的channel名称
     */
    String getChannelName() {
        return prefixName("redisson_lock__channel", getRawName());
    }


    /**
     * 加锁方法。
     */
    @Override
    public void lock() {
        try {
            lock(-1, null, false);
        } catch (InterruptedException e) {
            throw new IllegalStateException();
        }
    }

    @Override
    public void lock(long leaseTime, TimeUnit unit) {
        try {
            lock(leaseTime, unit, false);
        } catch (InterruptedException e) {
            throw new IllegalStateException();
        }
    }

    @Override
    public void lockInterruptibly() throws InterruptedException {
        lock(-1, null, true);
    }


    @Override
    public void lockInterruptibly(long leaseTime, TimeUnit unit) throws InterruptedException {
        lock(leaseTime, unit, true);
    }


    /**
     * 尝试获取锁。如果成功获取锁，则会根据提供的租约时间维护锁的生命周期，并在锁到期前尝试续订。
     * 如果锁已被其他线程持有，此方法将根据参数配置进入等待状态，直到获取锁或者遇到中断。
     * @param leaseTime     锁的租约时间
     * @param unit          租约时间的单位
     * @param interruptibly 是否可以被中断。如果为true，在等待锁时，如果线程被中断将抛出InterruptedException。
     * @throws InterruptedException 如果线程在等待锁时被中断且interruptibly为true。
     */
    private void lock(long leaseTime, TimeUnit unit, boolean interruptibly) throws InterruptedException {
        long threadId = Thread.currentThread().getId();
        Long ttl = tryAcquire(-1, leaseTime, unit, threadId);
        // 尝试获取锁，如果立即成功，则返回
        if (ttl == null) {
            return;
        }
        // 订阅
        CompletableFuture<RedissonLockEntry> future = subscribe(threadId);
        pubSub.timeout(future);
        RedissonLockEntry entry;
        if (interruptibly) {
            entry = commandExecutor.getInterrupted(future);
        } else {
            entry = commandExecutor.get(future);
        }

        try {
            // 循环尝试获取锁，直到成功或遇到其他终止条件
            while (true) {
                ttl = tryAcquire(-1, leaseTime, unit, threadId);
                // 成功获取锁
                if (ttl == null) {
                    break;
                }

                // 等待锁释放
                if (ttl >= 0) {
                    try {
                        entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    } catch (InterruptedException e) {
                        if (interruptibly) {
                            throw e;
                        }
                        entry.getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                    }
                } else {
                    if (interruptibly) {
                        entry.getLatch().acquire();
                    } else {
                        entry.getLatch().acquireUninterruptibly();
                    }
                }
            }
        } finally {
            // 无论是否成功获取锁，最后都取消订阅
            unsubscribe(entry, threadId);
        }
    }


    /**
     * 尝试获取资源的同步方法，是异步获取资源的同步封装。
     * @return 如果成功获取资源，返回资源的ID,否则返回null。
     */
    private Long tryAcquire(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        RFuture<Long> future = tryAcquireAsync0(waitTime, leaseTime, unit, threadId);
        return get(future);
    }


    /**
     * 尝试异步获取资源，带有等待时间和租约时间。
     * @return RFuture<Long> 返回一个包含获取成功后的资源计数的RFuture对象
     */
    private RFuture<Long> tryAcquireAsync0(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        return getServiceManager().execute(() -> tryAcquireAsync(waitTime, leaseTime, unit, threadId));
    }


    /**
     * 尝试异步获取锁，根据提供的等待时间和租用时间来尝试锁定。
     * @param waitTime 尝试获取锁的最大等待时间
     * @param leaseTime 锁的租用时间
     * @param unit 时间单位
     * @param threadId 线程ID，用于标识请求锁的线程
     * @return 一个未来对象，表示锁的剩余生存时间
     */
    private RFuture<Long> tryAcquireAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        RFuture<Long> ttlRemainingFuture;

        // 根据leaseTime的值决定使用内部锁租期还是用户指定的租期
        if (leaseTime > 0) {
            ttlRemainingFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_LONG);
        }
        else {
            ttlRemainingFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_LONG);
        }

        // 处理未同步的情况，并包装成CompletableFutureWrapper对象
        CompletionStage<Long> s = handleNoSync(threadId, ttlRemainingFuture);
        ttlRemainingFuture = new CompletableFutureWrapper<>(s);

        // 根据ttlRemaining的值来决定是否设置锁的过期时间或启动续期任务
        CompletionStage<Long> f = ttlRemainingFuture.thenApply(ttlRemaining -> {
            if (ttlRemaining == null) {
                if (leaseTime > 0) {
                    internalLockLeaseTime = unit.toMillis(leaseTime);
                } else {
                    scheduleExpirationRenewal(threadId);
                }
            }
            return ttlRemaining;
        });
        return new CompletableFutureWrapper<>(f);
    }



    /**
     * 尝试获取锁。
     * 该方法尝试异步获取锁，并立即返回一个布尔值，指示是否成功获取了锁。
     * 没有参数。
     *
     * @return boolean - 如果成功获取锁则返回true，否则返回false。
     */
    @Override
    public boolean tryLock() {
        return get(tryLockAsync());
    }

    /**
     * 尝试异步获取锁。
     * 该方法尝试为指定的线程ID获取锁。它以异步的方式执行，不阻塞当前线程，
     * 并返回一个未来对象，该对象表示锁是否成功获取。
     *
     * @param threadId 线程ID，标识尝试获取锁的线程。
     * @return RFuture<Boolean> 返回一个表示锁获取操作结果的未来对象。如果锁被成功获取，返回true；否则返回false。
     */
    @Override
    public RFuture<Boolean> tryLockAsync(long threadId) {
        return getServiceManager().execute(() -> tryAcquireOnceAsync(-1, -1, null, threadId));
    }


    /**
     * 尝试异步获取锁，一旦获取即处理锁的续期或设置过期时间。
     * @param waitTime  尝试获取锁的最大等待时间
     * @param leaseTime 锁的租期，如果为0则使用内部默认租期
     * @param unit      时间单位
     * @param threadId  请求锁的线程ID
     * @return 返回一个未来对象，表示是否成功获取锁
     */
    private RFuture<Boolean> tryAcquireOnceAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId) {
        CompletionStage<Boolean> acquiredFuture;
        // 加锁时是否设置过期时间
        // 1、未设置过期时间（-1）时则会有watchDog的锁续约，注册了加锁事件的续约任务
        if (leaseTime > 0) {
            acquiredFuture = tryLockInnerAsync(waitTime, leaseTime, unit, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
        } else {
            acquiredFuture = tryLockInnerAsync(waitTime, internalLockLeaseTime, TimeUnit.MILLISECONDS, threadId, RedisCommands.EVAL_NULL_BOOLEAN);
        }

        // 处理异步获取锁结果的同步问题
        acquiredFuture = handleNoSync(threadId, acquiredFuture);

        // 处理获取锁后的逻辑：如果锁被成功获取，根据是否指定了租期来设置锁的过期时间
        CompletionStage<Boolean> f = acquiredFuture.thenApply(acquired -> {
            // 成功获取锁的后续处理
            if (acquired) {
                if (leaseTime > 0) {
                    internalLockLeaseTime = unit.toMillis(leaseTime);
                } else {
                    scheduleExpirationRenewal(threadId); // 调整锁的过期时间
                }
            }
            return acquired;
        });
        return new CompletableFutureWrapper<>(f);
    }

    /**
     * 尝试以异步方式获取锁。该方法会检查锁是否已被其他线程持有，如果没有被持有，则会尝试获取锁，并设置锁的过期时间。
     * 如果锁已被持有，方法将返回当前锁的剩余时间。
     * @param waitTime  等待获取锁的最大时间。
     * @param leaseTime 锁的持有时间。
     * @param unit      时间单位。
     * @param threadId  线程标识符，用于生成唯一的锁名称。
     * @param command   Redis命令，用于执行具体的锁逻辑。
     * @return 如果成功获取锁，则返回null; 如果锁已被持有，返回锁的剩余时间（毫秒）。
     */
    <T> RFuture<T> tryLockInnerAsync(long waitTime, long leaseTime, TimeUnit unit, long threadId, RedisStrictCommand<T> command) {
        /*
            // 参数举例
            //  KEYS[1] = redis_key_name
            //  ARGV[1] = 1000
            //  ARGV[2] = 4defdd2794df:12

             if ((redis.call('exists',KEYS[1]) == 0) or (redis.call('hexists',KEYS[1],ARGV[2])==1)) then
                 redis.call('hincrby',key, threadId, 1);
                 redis.call('pexpire',key, leaseTime);
                 return nil;
             end;
             return redis.call('pttl',key)
         */

        return commandExecutor.syncedEval(getRawName(), LongCodec.INSTANCE, command,
                "if ((redis.call('exists', KEYS[1]) == 0) " +
                        "or (redis.call('hexists', KEYS[1], ARGV[2]) == 1)) then " +
                        "redis.call('hincrby', KEYS[1], ARGV[2], 1); " +
                        "redis.call('pexpire', KEYS[1], ARGV[1]); " +
                        "return nil; " +
                        "end; " +
                        "return redis.call('pttl', KEYS[1]);",
                Collections.singletonList(getRawName()), unit.toMillis(leaseTime), getLockName(threadId));
    }

    /**
     * 加锁方法
     * 在waitTime时间范围内尝试获取锁,如果获取到锁,则设置过期时间leaseTime
     */
    @Override
    public boolean tryLock(long waitTime, long leaseTime, TimeUnit unit) throws InterruptedException {
        long time = unit.toMillis(waitTime);
        long current = System.currentTimeMillis();
        long threadId = Thread.currentThread().getId();

        // 第一步：尝试获取锁
        Long ttl = tryAcquire(waitTime, leaseTime, unit, threadId);
        // ttl为空,说明获取到锁了,直接返回响应结果即可
        if (ttl == null) {
            return true;
        }

        // 已经超过了获取锁的等待时间
        time -= System.currentTimeMillis() - current;
        if (time <= 0) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }

        // 第二步: 订阅解锁消息通知
        current = System.currentTimeMillis();
        CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(threadId);
        try {
            subscribeFuture.get(time, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            if (!subscribeFuture.completeExceptionally(new RedisTimeoutException(
                    "Unable to acquire subscription lock after " + time + "ms. " +
                            "Try to increase 'subscriptionsPerConnection' and/or 'subscriptionConnectionPoolSize' parameters."))) {
                subscribeFuture.whenComplete((res, ex) -> {
                    if (ex == null) {
                        unsubscribe(res, threadId);
                    }
                });
            }
            acquireFailed(waitTime, unit, threadId);
            return false;
        } catch (ExecutionException e) {
            acquireFailed(waitTime, unit, threadId);
            return false;
        }



        try {
            // 已经超过了获取锁的等待时间
            time -= System.currentTimeMillis() - current;
            if (time <= 0) {
                acquireFailed(waitTime, unit, threadId);
                return false;
            }

            while (true) {
                // 1、尝试获取锁
                long currentTime = System.currentTimeMillis();
                ttl = tryAcquire(waitTime, leaseTime, unit, threadId);

                // 2、获取到锁,return true
                if (ttl == null) {
                    return true;
                }

                // 3、等待超时,return false
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(waitTime, unit, threadId);
                    return false;
                }

                // 4、
                currentTime = System.currentTimeMillis();
                if (ttl >= 0 && ttl < time) {
                    commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(ttl, TimeUnit.MILLISECONDS);
                } else {
                    commandExecutor.getNow(subscribeFuture).getLatch().tryAcquire(time, TimeUnit.MILLISECONDS);
                }

                // 5、等待超时,return false
                time -= System.currentTimeMillis() - currentTime;
                if (time <= 0) {
                    acquireFailed(waitTime, unit, threadId);
                    return false;
                }
            }
        } finally {
            // 取消订阅
            unsubscribe(commandExecutor.getNow(subscribeFuture), threadId);
        }
    }

    protected CompletableFuture<RedissonLockEntry> subscribe(long threadId) {
        return pubSub.subscribe(getEntryName(), getChannelName());
    }

    protected void unsubscribe(RedissonLockEntry entry, long threadId) {
        pubSub.unsubscribe(entry, getEntryName(), getChannelName());
    }

    @Override
    public boolean tryLock(long waitTime, TimeUnit unit) throws InterruptedException {
        return tryLock(waitTime, -1, unit);
    }

    @Override
    protected void cancelExpirationRenewal(Long threadId) {
        super.cancelExpirationRenewal(threadId);
        this.internalLockLeaseTime = getServiceManager().getCfg().getLockWatchdogTimeout();
    }

    @Override
    public RFuture<Boolean> forceUnlockAsync() {
        cancelExpirationRenewal(null);
        return commandExecutor.syncedEvalWithRetry(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('del', KEYS[1]) == 1) then "
                        + "redis.call(ARGV[2], KEYS[2], ARGV[1]); "
                        + "return 1 "
                        + "else "
                        + "return 0 "
                        + "end",
                Arrays.asList(getRawName(), getChannelName()), LockPubSub.UNLOCK_MESSAGE, getSubscribeService().getPublishCommand());
    }


    /**
     * 异步解锁内部方法。
     * 该方法使用Redis的EVAL命令来执行解锁逻辑。如果锁存在且属于调用线程，则减少锁的计数器。如果计数器大于0，则更新锁的过期时间；
     * 如果计数器等于0，则删除锁，并发布解锁消息。
     *
     * @param threadId 线程ID，用于标识哪个线程持有的锁。
     * @return RFuture<Boolean> 返回一个未来对象，表示操作是否成功。如果成功解锁返回true，否则返回false。
     */
    protected RFuture<Boolean> unlockInnerAsync(long threadId) {
        /*
          // 没有上锁,返回return null
          if (redis.call('hexists', key, threadId) == 0) then')
          return nil;
          end;

          // 删除该线程的锁
          local counter = redis.call('hincrby', key, threadId, -1);
          if (counter > 0) then
               redis.call('pexpire', key, internalLockLeaseTime)  // 覆盖超时时间
               return 0;
          else
               redis.call('del',key);
               redis.call('PUBLISH',redisson_lock__channel:{key});
               return 1;
          end
          return nil;
         */
        return evalWriteAsync(getRawName(), LongCodec.INSTANCE, RedisCommands.EVAL_BOOLEAN,
                "if (redis.call('hexists', KEYS[1], ARGV[3]) == 0) then " +
                        "return nil;" +
                        "end; " +
                        "local counter = redis.call('hincrby', KEYS[1], ARGV[3], -1); " +
                        "if (counter > 0) then " +
                        "redis.call('pexpire', KEYS[1], ARGV[2]); " +
                        "return 0; " +
                        "else " +
                        "redis.call('del', KEYS[1]); " +
                        "redis.call(ARGV[4], KEYS[2], ARGV[1]); " +
                        "return 1; " +
                        "end; " +
                        "return nil;",
                Arrays.asList(getRawName(), getChannelName()),
                // channel发送消息的类别，此处解锁为0
                LockPubSub.UNLOCK_MESSAGE,
                // watchDog配置的超时时间，默认为30s
                internalLockLeaseTime,
                // 这里的lockName指的是uuid和threadId组合的唯一值
                getLockName(threadId),
                getSubscribeService().getPublishCommand());
    }

    @Override
    public RFuture<Void> lockAsync(long leaseTime, TimeUnit unit, long currentThreadId) {
        CompletableFuture<Void> result = new CompletableFuture<>();
        RFuture<Long> ttlFuture = tryAcquireAsync0(-1, leaseTime, unit, currentThreadId);
        ttlFuture.whenComplete((ttl, e) -> {
            if (e != null) {
                result.completeExceptionally(e);
                return;
            }
            // lock acquired
            if (ttl == null) {
                if (!result.complete(null)) {
                    unlockAsync(currentThreadId);
                }
                return;
            }
            CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(currentThreadId);
            pubSub.timeout(subscribeFuture);
            subscribeFuture.whenComplete((res, ex) -> {
                if (ex != null) {
                    result.completeExceptionally(ex);
                    return;
                }

                lockAsync(leaseTime, unit, res, result, currentThreadId);
            });
        });

        return new CompletableFutureWrapper<>(result);
    }

    private void lockAsync(long leaseTime, TimeUnit unit,
                           RedissonLockEntry entry, CompletableFuture<Void> result, long currentThreadId) {
        RFuture<Long> ttlFuture = tryAcquireAsync0(-1, leaseTime, unit, currentThreadId);
        ttlFuture.whenComplete((ttl, e) -> {
            if (e != null) {
                unsubscribe(entry, currentThreadId);
                result.completeExceptionally(e);
                return;
            }

            // lock acquired
            if (ttl == null) {
                unsubscribe(entry, currentThreadId);
                if (!result.complete(null)) {
                    unlockAsync(currentThreadId);
                }
                return;
            }

            if (entry.getLatch().tryAcquire()) {
                lockAsync(leaseTime, unit, entry, result, currentThreadId);
            } else {
                // waiting for message
                AtomicReference<Timeout> futureRef = new AtomicReference<>();
                Runnable listener = () -> {
                    if (futureRef.get() != null) {
                        futureRef.get().cancel();
                    }
                    lockAsync(leaseTime, unit, entry, result, currentThreadId);
                };

                entry.addListener(listener);

                if (ttl >= 0) {
                    Timeout scheduledFuture = getServiceManager().newTimeout(timeout -> {
                        if (entry.removeListener(listener)) {
                            lockAsync(leaseTime, unit, entry, result, currentThreadId);
                        }
                    }, ttl, TimeUnit.MILLISECONDS);
                    futureRef.set(scheduledFuture);
                }
            }
        });
    }




    /**
     * 尝试异步获取锁，如果锁不可用，则根据指定的等待时间和租用时间尝试获取锁。
     * @param waitTime        等待获取锁的最大时间
     * @param leaseTime       锁的租用时间
     * @param unit            时间单位
     * @param currentThreadId 当前线程ID
     * @return 返回一个CompletableFuture，如果成功获取锁则完成为true，否则为false
     */
    @Override
    public RFuture<Boolean> tryLockAsync(long waitTime, long leaseTime, TimeUnit unit,
                                         long currentThreadId) {
        CompletableFuture<Boolean> result = new CompletableFuture<>();

        // 将等待时间转换为毫秒，并初始化当前时间
        AtomicLong time = new AtomicLong(unit.toMillis(waitTime));
        long currentTime = System.currentTimeMillis();

        // 尝试异步获取锁
        RFuture<Long> ttlFuture = tryAcquireAsync0(waitTime, leaseTime, unit, currentThreadId);

        // 处理获取锁的结果
        ttlFuture.whenComplete((ttl, e) -> {
            // 如果过程中有异常，则将异常传递给result
            if (e != null) {
                result.completeExceptionally(e);
                return;
            }

            // 如果成功获取锁
            if (ttl == null) {
                // 如果result未完成，则完成为true，并释放锁
                if (!result.complete(true)) {
                    unlockAsync(currentThreadId);
                }
                return;
            }

            // 计算已过去的时间，并更新剩余等待时间
            long el = System.currentTimeMillis() - currentTime;
            time.addAndGet(-el);

            // 如果剩余等待时间小于等于0，则尝试失败
            if (time.get() <= 0) {
                trySuccessFalse(currentThreadId, result);
                return;
            }

            // 订阅锁释放事件，并设置超时
            long current = System.currentTimeMillis();
            AtomicReference<Timeout> futureRef = new AtomicReference<>();
            CompletableFuture<RedissonLockEntry> subscribeFuture = subscribe(currentThreadId);
            pubSub.timeout(subscribeFuture, time.get());
            subscribeFuture.whenComplete((r, ex) -> {
                // 如果过程中有异常，则将异常传递给result
                if (ex != null) {
                    result.completeExceptionally(ex);
                    return;
                }

                // 取消之前的定时任务
                if (futureRef.get() != null) {
                    futureRef.get().cancel();
                }

                // 更新剩余时间
                long elapsed = System.currentTimeMillis() - current;
                time.addAndGet(-elapsed);

                // 递归尝试获取锁
                tryLockAsync(time, waitTime, leaseTime, unit, r, result, currentThreadId);
            });

            // 如果订阅未完成，则设置一个定时任务，如果订阅在设定时间内未完成，则视为获取锁失败
            if (!subscribeFuture.isDone()) {
                Timeout scheduledFuture = getServiceManager().newTimeout(timeout -> {
                    if (!subscribeFuture.isDone()) {
                        subscribeFuture.cancel(false);
                        trySuccessFalse(currentThreadId, result);
                    }
                }, time.get(), TimeUnit.MILLISECONDS);
                futureRef.set(scheduledFuture);
            }
        });

        // 返回包装后的CompletableFuture
        return new CompletableFutureWrapper<>(result);
    }


    /**
     * 尝试异步获取锁。
     *
     * @param time            锁的剩余时间，以毫秒为单位。
     * @param waitTime        尝试获取锁的最大等待时间。
     * @param leaseTime       锁的租期，即持有锁的时间。
     * @param unit            时间单位。
     * @param entry           锁的入口标识，用于多线程同步。
     * @param result          获取锁操作的结果，通过CompletableFuture完成。
     * @param currentThreadId 当前线程的ID。
     */
    private void tryLockAsync(AtomicLong time, long waitTime, long leaseTime, TimeUnit unit,
                              RedissonLockEntry entry, CompletableFuture<Boolean> result, long currentThreadId) {
        // 如果结果已完成，则取消订阅并返回
        if (result.isDone()) {
            unsubscribe(entry, currentThreadId);
            return;
        }

        // 如果锁的剩余时间小于等于0，则取消订阅，并将结果标记为失败
        if (time.get() <= 0) {
            unsubscribe(entry, currentThreadId);
            trySuccessFalse(currentThreadId, result);
            return;
        }

        // 记录当前时间
        long curr = System.currentTimeMillis();
        // 尝试异步获取锁
        RFuture<Long> ttlFuture = tryAcquireAsync0(waitTime, leaseTime, unit, currentThreadId);
        ttlFuture.whenComplete((ttl, e) -> {
            // 如果遇到异常，则取消订阅，并将异常传递给结果
            if (e != null) {
                unsubscribe(entry, currentThreadId);
                result.completeExceptionally(e);
                return;
            }

            // 如果成功获取锁（ttl为null），则取消订阅，并将结果标记为成功
            if (ttl == null) {
                unsubscribe(entry, currentThreadId);
                if (!result.complete(true)) {
                    unlockAsync(currentThreadId);
                }
                return;
            }

            // 更新锁的剩余时间，并检查是否需要再次尝试获取锁
            long el = System.currentTimeMillis() - curr;
            time.addAndGet(-el);

            // 如果经过时间超过等待时间，则取消订阅，并将结果标记为失败
            if (time.get() <= 0) {
                unsubscribe(entry, currentThreadId);
                trySuccessFalse(currentThreadId, result);
                return;
            }

            // 如果可以立即获取锁，则再次尝试获取锁
            long current = System.currentTimeMillis();
            if (entry.getLatch().tryAcquire()) {
                tryLockAsync(time, waitTime, leaseTime, unit, entry, result, currentThreadId);
            } else {
                // 如果不能立即获取锁，则注册监听器并延迟再次尝试
                AtomicBoolean executed = new AtomicBoolean();
                AtomicReference<Timeout> futureRef = new AtomicReference<>();
                Runnable listener = () -> {
                    executed.set(true);
                    if (futureRef.get() != null) {
                        futureRef.get().cancel();
                    }

                    long elapsed = System.currentTimeMillis() - current;
                    time.addAndGet(-elapsed);

                    tryLockAsync(time, waitTime, leaseTime, unit, entry, result, currentThreadId);
                };
                entry.addListener(listener);

                long t = time.get();
                // 根据锁的剩余时间调整等待时间
                if (ttl >= 0 && ttl < time.get()) {
                    t = ttl;
                }
                // 如果监听器还未被触发，则安排延迟执行
                if (!executed.get()) {
                    Timeout scheduledFuture = getServiceManager().newTimeout(timeout -> {
                        if (entry.removeListener(listener)) {
                            long elapsed = System.currentTimeMillis() - current;
                            time.addAndGet(-elapsed);

                            tryLockAsync(time, waitTime, leaseTime, unit, entry, result, currentThreadId);
                        }
                    }, t, TimeUnit.MILLISECONDS);
                    futureRef.set(scheduledFuture);
                }
            }
        });
    }

}
